from queue import Queue, Empty
from threading import Thread
from enum import Enum
from abc import ABCMeta, abstractmethod
import asyncio


class _AsyncSessionMeta(ABCMeta):
    def __init__(cls, name, bases, attrs):  # 定义新类的时候执行
        super().__init__(name, bases, attrs)
        cls._session_dict: dict[float, 'AsyncSession'] = {}
        cls._thread = None

    async def main(cls):  # 新类的类方法
        while True:
            for session in cls._session_dict.values():
                session.create_task()
            await asyncio.sleep(0.1)

    def get_session(cls, key):
        if cls._thread is None:
            cls._thread = Thread(target=asyncio.run, args=(cls.main(),), daemon=True)
            cls._thread.start()
        if key not in cls._session_dict:
            cls._session_dict[key] = cls(key)
        return cls._session_dict[key]


class AsyncSession(metaclass=_AsyncSessionMeta):
    class State(Enum):
        CREATE = 0  # 刚创建，还没有启动task
        LIVE = 1  # 已经启动task
        DIE = 2  # 已超时死亡

    def __init__(self, key):
        self._key = key
        self._input_queue = Queue()
        self._output_queue = Queue()
        self._state = AsyncSession.State.CREATE

    def consume(self, p):
        self._input_queue.put(p)
        try:
            return self._output_queue.get(timeout=10)
        except Empty:
            self.die()
            return None

    @abstractmethod
    async def produce(self):
        pass

    def die(self):
        self._state = AsyncSession.State.DIE
        if self._key in self._session_dict:
            del self._session_dict[self._key]

    async def __die_timer(self):
        await asyncio.sleep(24 * 60 * 60)
        self.die()

    def create_task(self):
        if self._state != AsyncSession.State.CREATE:
            return
        self._state = AsyncSession.State.LIVE
        asyncio.create_task(self.produce())
        asyncio.create_task(self.__die_timer())
